package mr

import "C"
import (
	"log"
	"net"
	"net/http"
	"net/rpc"
	"os"
	"strconv"
	"sync"
	"time"
)

//coordinator need to issue files to different workers
//A slice for keep status of workers
//A slice for keep status of tasks
//first issue map tasks to workers
//when workers RPC server, they will get task if they can
//if someone crashed so that over 10s after get task,
//Coordinator should resume it crashed and re-issue the task to other worker
//when all map tasks has been finished, start issue reduce tasks to worker
//when both map and reduce has been complete finished, issue "plase exit" to workers so that workers can exit
//when here is no worker Coordinator change ret to ture so that it can exit

type Coordinator struct {
	// Your definitions here.
	Filenames []string
	NReduce   int
	//nums of files
	MapTasks []Task
	//nums of nReduce
	ReduceTasks  []Task
	RunningTasks []Task
	//workers, exit when length of workers is zero
	WorkerStatuses []WorkerStatus

	FinishedTasks []Task
}

type Task struct {
	//map/reduce/none/exit
	TaskType string
	//map task
	NReduce  int
	Filename string
	//index of task
	TaskIndex int
	//StartTime in Running Tasks if over 10s timeout reissue it
	StartTime time.Time
	WorkId    int
}

type WorkerStatus struct {
	//hearten time
	//if over 10s timeout remove it from workers
	LastHearten time.Time
	WorkerId    int
}

var mu sync.Mutex

// Your code here -- RPC handlers for the worker to call.
func (c *Coordinator) Hearten(args *AskingArgs, reply *Reply) error {
	mu.Lock()
	defer mu.Unlock()
	reply.NReduce = c.NReduce
	log.Println("worker" + strconv.Itoa(args.WorkId) + "发送心跳，请求类型为" + args.AskingType)
	//更新Worker中的心跳时间
	hasWorker := false
	for _, worker := range c.WorkerStatuses {
		if worker.WorkerId == args.WorkId {
			worker.LastHearten = time.Now()
			hasWorker = true
			break
		}
	}
	if !hasWorker {
		worker := WorkerStatus{}
		worker.WorkerId = args.WorkId
		worker.LastHearten = time.Now()
		c.WorkerStatuses = append(c.WorkerStatuses, worker)
	}

	//如果worker在工作中就继续
	if args.AskingType == "working" {
		return nil
	}

	//如果是完成任务,从RunningTask中移到FinishedTask
	if args.AskingType == "finish" {
		log.Println("Worker" + strconv.Itoa(args.WorkId) + "完成任务" + args.TaskType + strconv.Itoa(args.TaskIndex))
		for i, task := range c.RunningTasks {
			if task.TaskType == args.TaskType && task.TaskIndex == args.TaskIndex {
				log.Println("找到运行中的该任务将其移入finish中" + task.TaskType + strconv.Itoa(task.TaskIndex))
				c.FinishedTasks = append(c.FinishedTasks, c.RunningTasks[i])
				copy(c.RunningTasks[i:], c.RunningTasks[i+1:])
				c.RunningTasks = c.RunningTasks[:len(c.RunningTasks)-1]
				break
			}
		}
		log.Println("剩余运行中任务")
		for _, runningTask := range c.RunningTasks {
			log.Println("running任务类型" + runningTask.TaskType + "，任务index：" + strconv.Itoa(runningTask.TaskIndex))
		}
		return nil
	}

	//如果请求任务就分配
	//先检查Map任务是否全部完成
	//如果还有Map任务先把Map任务分配出去并将MapTask移到RunningTask
	if len(c.MapTasks) > 0 {
		reply.TaskType = "map"
		reply.TaskIndex = c.MapTasks[0].TaskIndex
		reply.StartTime = time.Now()
		reply.Filename = c.MapTasks[0].Filename

		log.Println("Worker" + strconv.Itoa(args.WorkId) + "添加runningtask" + c.MapTasks[0].TaskType + strconv.Itoa(c.MapTasks[0].TaskIndex))
		c.MapTasks[0].StartTime = time.Now()
		c.MapTasks[0].WorkId = args.WorkId
		c.RunningTasks = append(c.RunningTasks, c.MapTasks[0])

		log.Println("Worker" + strconv.Itoa(args.WorkId) + "剩余运行中任务")
		for _, runningTask := range c.RunningTasks {
			log.Println("Worker" + strconv.Itoa(args.WorkId) + "running任务类型" + runningTask.TaskType + "，任务index：" + strconv.Itoa(runningTask.TaskIndex))
		}

		log.Println("分配map任务id：" + strconv.Itoa(reply.TaskIndex) + "给Worker" + strconv.Itoa(args.WorkId) + "，文件名为：" + reply.Filename)
		copy(c.MapTasks[:len(c.MapTasks)-1], c.MapTasks[1:])
		c.MapTasks = c.MapTasks[:len(c.MapTasks)-1]

		log.Println("Worker" + strconv.Itoa(args.WorkId) + "剩余map任务")
		for _, mapTask := range c.MapTasks {
			log.Println("Worker" + strconv.Itoa(args.WorkId) + "map 任务index" + strconv.Itoa(mapTask.TaskIndex) + "，文件名为：" + mapTask.Filename)
		}

		return nil
	}
	//如果还有map task正在运行也不能分配reduce task
	hasMap := false
	for _, task := range c.RunningTasks {
		if task.TaskType == "map" {
			hasMap = true
			break
		}
	}
	if hasMap {
		reply.TaskType = "none"
		return nil
	}

	//再检查Reduce任务是否完成
	//如果还有Reduce任务先把Reduce任务分配出去并将ReduceTask移到RunningTask
	if len(c.ReduceTasks) > 0 {
		reply.TaskType = "reduce"
		reply.TaskIndex = c.ReduceTasks[0].TaskIndex
		reply.StartTime = time.Now()

		log.Println("Worker" + strconv.Itoa(args.WorkId) + "添加运行任务：" + c.ReduceTasks[0].TaskType + strconv.Itoa(c.ReduceTasks[0].TaskIndex))
		c.ReduceTasks[0].StartTime = time.Now()
		c.ReduceTasks[0].WorkId = args.WorkId
		c.RunningTasks = append(c.RunningTasks, c.ReduceTasks[0])

		log.Println("Worker" + strconv.Itoa(args.WorkId) + "运行中任务")
		for _, runningTask := range c.RunningTasks {
			log.Println("Worker" + strconv.Itoa(args.WorkId) + "running任务类型" + runningTask.TaskType + "，任务index：" + strconv.Itoa(runningTask.TaskIndex))
		}

		log.Println("Worker" + strconv.Itoa(args.WorkId) + "分配reduce任务id：" + strconv.Itoa(reply.TaskIndex))
		copy(c.ReduceTasks[:len(c.ReduceTasks)-1], c.ReduceTasks[1:])
		c.ReduceTasks = c.ReduceTasks[:len(c.ReduceTasks)-1]

		log.Println("Worker" + strconv.Itoa(args.WorkId) + "剩余reduce任务")
		for _, reduceTask := range c.ReduceTasks {
			log.Println("Worker" + strconv.Itoa(args.WorkId) + "reduce 任务index" + strconv.Itoa(reduceTask.TaskIndex))
		}

		return nil
	}

	//如果只有运行中的任务就让worker待定
	if len(c.MapTasks) > 0 || len(c.ReduceTasks) > 0 || len(c.RunningTasks) > 0 {
		reply.TaskType = "none"
		return nil
	}

	//如果Map与Reduce任务都完成了就返回please exit任务，并从ruunning中移除
	reply.TaskType = "exit"
	for i, worker := range c.WorkerStatuses {
		if worker.WorkerId == args.WorkId {
			copy(c.WorkerStatuses[i:], c.WorkerStatuses[i+1:])
			c.WorkerStatuses = c.WorkerStatuses[:len(c.WorkerStatuses)-1]
		}
	}
	return nil
}

//
// start a thread that listens for RPCs from worker.go
//
func (c *Coordinator) server() {
	rpc.Register(c)
	rpc.HandleHTTP()
	//l, e := net.Listen("tcp", ":1234")
	sockname := coordinatorSock()
	log.Println("sockname is " + sockname)
	os.Remove(sockname)
	l, e := net.Listen("unix", sockname)
	if e != nil {
		log.Fatal("listen error:", e)
	}
	go http.Serve(l, nil)
}

//
// main/mrcoordinator.go calls Done() periodically to find out
// if the entire job has finished.
//
func (c *Coordinator) Done() bool {
	ret := false

	// Your code here.
	//if job had finished, ret = true
	if len(c.RunningTasks) == 0 &&
		len(c.MapTasks) == 0 &&
		len(c.ReduceTasks) == 0 &&
		len(c.WorkerStatuses) == 0 {
		log.Println("所有任务处理完毕")
		ret = true
	}
	return ret
}

//
// create a Coordinator.
// main/mrcoordinator.go calls this function.
// nReduce is the number of reduce tasks to use.
//
func MakeCoordinator(files []string, nReduce int) *Coordinator {
	c := Coordinator{}

	// Your code here.
	c.NReduce = nReduce
	c.Filenames = files

	for i, file := range files {
		task := Task{}
		task.TaskType = "map"
		task.Filename = file
		task.NReduce = nReduce
		task.TaskIndex = i
		c.MapTasks = append(c.MapTasks, task)
		log.Println("添加map任务：i=" + strconv.Itoa(i) + ", filename=" + file)
	}

	for i := 0; i < nReduce; i++ {
		task := Task{}
		task.TaskType = "reduce"
		task.TaskIndex = i
		task.NReduce = nReduce
		c.ReduceTasks = append(c.ReduceTasks, task)
		log.Println("添加reduce任务：i=" + strconv.Itoa(i))
	}

	c.server()

	//开启一个线程每秒检查有没有超时任务
	//检查有没有失效worker
	go func() {
		for {
			time.Sleep(time.Second / 10)
			for i, task := range c.RunningTasks {
				//由于性能问题10s可能过短
				if time.Now().Sub(task.StartTime).Seconds() >= 10 {
					log.Println("发现超时10s任务，超时时间" + strconv.Itoa(int(time.Now().Sub(task.StartTime).Seconds())))
					if task.TaskType == "map" {
						log.Println("把超时任务放回Map任务")
						c.MapTasks = append(c.MapTasks, task)
					} else if task.TaskType == "reduce" {
						log.Println("把超时任务放回Reduce任务")
						c.ReduceTasks = append(c.ReduceTasks, task)
					}
					log.Println("把超时任务删除")
					copy(c.RunningTasks[i:], c.RunningTasks[i+1:])
					c.RunningTasks = c.RunningTasks[:len(c.RunningTasks)-1]
					break
				}
			}
			for i, status := range c.WorkerStatuses {
				//由于性能问题10s可能过短
				if time.Now().Sub(status.LastHearten).Seconds() >= 15 {
					log.Println("发现超时15s Worker，超时时间" + strconv.Itoa(int(time.Now().Sub(status.LastHearten).Seconds())))
					log.Println("把crash worker删除")
					copy(c.WorkerStatuses[i:], c.WorkerStatuses[i+1:])
					c.WorkerStatuses = c.WorkerStatuses[:len(c.WorkerStatuses)-1]
					log.Println("剩余worker")
					for _, workerStatus := range c.WorkerStatuses {
						log.Println("worker" + strconv.Itoa(workerStatus.WorkerId) + "上一次心跳" + workerStatus.LastHearten.String())
					}
					break
				}
			}
		}
	}()
	return &c
}
